
package com.service.socket.rabbitmq.listener;

import com.alibaba.fastjson.JSON;
import com.service.dao.enums.KlineTypeEnum;
import com.service.dao.model.CoinsPrice;
import com.service.dao.model.TradingSpotTrades;
import com.service.match.common.entity.Depth;
import com.service.match.common.entity.PushDepth;
import com.service.match.common.entity.SpotOrderVo;
import com.service.match.common.entity.SpotTradeVo;
import com.service.match.common.enums.TradingSpotEnum;
import com.service.match.common.utils.TradingSpotUtils;
import com.service.socket.common.entity.SendMessageVo;
import com.service.socket.common.entity.SingleMsgVo;
import com.service.socket.common.enums.MsgSendType;
import com.service.socket.rabbitmq.binding.MatchSink;
import com.service.socket.service.CoinsPriceService;
import com.service.socket.service.TradingKlineService;
import com.service.socket.service.TradingSpotPairsService;
import com.service.socket.service.TradingSpotTradesService;
import com.service.socket.support.KlineLockUtils;
import com.service.socket.support.TradingSpotCache;
import com.service.socket.ws.TradingSpotWS;
import com.service.socket.ws.service.KlineBroadcastService;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.Arrays;
import java.util.Date;

/**
 * @author flying-cattle
 * @ClassName: MatchOutResources
 * @Description: 撮合输出数据源
 * @date 2019年12月20日
 */

@Slf4j
@Component
@EnableBinding(MatchSink.class)
public class MatchReceiverListener {
    @Autowired
    private TradingSpotWS spotWS;
    @Autowired
    private TradingSpotCache spotCache;
    @Autowired
    private KlineLockUtils klineLockUtils;
    @Autowired
    private CoinsPriceService coinsPriceService;
    @Autowired
    private KlineBroadcastService broadcastService;
    @Autowired
    private TradingKlineService tradingKlineService;
    @Autowired
    private TradingSpotTradesService tradingSpotTradesService;

    @Getter
    private String depthStr = "";

    private SpotTradeVo lastTrade;

    /**
     * @return void 返回类型
     * @throws
     * @Title: push_depth
     * @Description: TODO(盘口深度数据)
     */
    @StreamListener(MatchSink.IN_PUSH_DEPTH)
    public void receiveDepth(String message) {
        if (StringUtils.isBlank(message)) {
            return;
        }
        PushDepth pd = JSON.parseObject(message, PushDepth.class);
        log.info("receive {} depth", pd.getSymbol());
        this.lastTrade = tradingSpotTradesService.getLastTradeVo(pd.getSymbol());
        if (ObjectUtils.isEmpty(this.lastTrade)) {
            CoinsPrice coinsPrice = coinsPriceService.getLastCoinPriceByCoinName(pd.getSymbol().split("/")[0]);
            this.lastTrade = new SpotTradeVo();
            this.lastTrade.setOrderPrice(coinsPrice.getStandardPrice());
            this.lastTrade.setTradeType(TradingSpotEnum.TradeType.BUY.getCode());
        }
        Depth depth = new Depth();
        depth.setPrice(this.lastTrade.getOrderPrice());
        depth.setIsBuy(TradingSpotUtils.isBuy(this.lastTrade.getTradeType()));
        pd.setDepth(depth);
        depthStr = JSON.toJSONString(pd);
        try {
            spotCache.cacheDepth(pd.getSymbol(), depthStr);
            spotWS.broadcastThemeMsg(MsgSendType.handicap,
                    SendMessageVo.builder().symbol(pd.getSymbol()).data(depthStr).build());
        } catch (Exception e) {
            log.error("socket 推送 handicap 错误:{}", e.getMessage());
            e.printStackTrace();
        }

    }

    /**
     * @return void 返回类型
     * @throws
     * @Title: push_depth
     * @Description: TODO(订单变化)
     */
    @StreamListener(MatchSink.IN_ALTER_SPOT_LIST)
    public void alterSpotList(SingleMsgVo singleMsgVo) {
        log.info("alterSpotList:{}", singleMsgVo);
        //todo：更新交易列表
        spotWS.sendSingleMsg(singleMsgVo.getMsgSendType(), singleMsgVo.getUserAccount(), singleMsgVo.getSendMessageVo());
    }

    @StreamListener(MatchSink.IN_KLINE_ITEM)
    public void receiveKlineItem(TradingSpotTrades trades) {
        log.info("kline receive trade:{}", trades);
        Arrays.stream(KlineTypeEnum.values()).forEach(item -> {
            String lockKey = KlineLockUtils.buildLock(item.getValue());
            if (klineLockUtils.lock(lockKey)) {
                try {
                    tradingKlineService.handlerTrade(trades, item);
                } finally {
                    klineLockUtils.unlock(lockKey);
                }
                //推送Kline
                broadcastService.broadcastKline(trades.getSymbol(), new Date());
            }
        });
        broadcastService.broadcastKlineStatistic(trades.getSymbol(), new Date());

    }

}
